package com.atguigu.flink.streamapi.trans;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/1/13
 * 
 *  按照传感器的id，进行vc的累加
 */
public class Demo8_ProcessKeyBY
{
    public static void main(String[] args) {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
                env
                   .socketTextStream("hadoop103", 8888)
                   .map(new WaterSensorMapFunction())
                   .keyBy(WaterSensor::getId)
                   //KeyedProcessFunction<K 分组key的类型, I 输入的类型, O 输出的类型>
                   .process(new KeyedProcessFunction<String, WaterSensor, String>()
                   {
                       //声明一个变量，用于保存之前累加的vc。 成为状态。累加操作也称为有状态的计算。
                       int sumVc = 0;
                       /*
                            WaterSensor value: 输入的数据
                            Context ctx： 环境，上下文。从这里获取一些工具。
                            Collector<String> out： 向下游输出的收集器
                        */
                       @Override
                       public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {

                           sumVc += value.getVc();

                           out.collect(ctx.getCurrentKey()+" 总水位:"+sumVc);

                       }
                   })
                   .print();
        
                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
    }
}
